package com.share.example.json2flat.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONPath;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.share.common.utils.GroovyUtil;
import com.share.example.json2flat.constants.ConfConstant;
import com.share.example.json2flat.mapper.MappingConfMapper;
import com.share.example.json2flat.model.db.MappingConf;
import com.share.example.json2flat.service.IMappingConfService;
import groovy.lang.GroovyObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author author
 * @since 2024-05-08
 */
@Slf4j
@Service
public class MappingConfServiceImpl extends ServiceImpl<MappingConfMapper, MappingConf> implements IMappingConfService {
    @Autowired
    private KafkaTemplate<Object, Object> template;
    //key=sourceflag
    private Map<String, List<MappingConf>> sourceMap = new HashMap<>();

    //key=sourceflag+fieldname
    private Map<String, GroovyObject> groovyMap = new HashMap<>();

    @Autowired
    private MappingConfMapper mappingConfMapper;

    @Override
    public Object init() {
        //1.定义变量
        Map<String, List<MappingConf>> _sourceMap = new HashMap<>();
        Map<String, GroovyObject> _groovyMap = new HashMap<>();

        //2.查询数据
        LambdaQueryWrapper<MappingConf> queryWrapper = new LambdaQueryWrapper();
        queryWrapper.eq(MappingConf::getStatus, "1");
        List<MappingConf> lists = mappingConfMapper.selectList(queryWrapper);
        for (MappingConf po : lists) {
            List list = _sourceMap.get(po.getSourceFlag());
            if (list == null) {
                list = new LinkedList();
                _sourceMap.put(po.getSourceFlag(), list);
            }
            list.add(po);

            //groovy
            if ("0".equals(po.getDealWay())) {
                try {
                    GroovyObject groovyObject = GroovyUtil.createGroovyObject(po.getGroovyScripy());
                    _groovyMap.put(po.getSourceFlag() + "-" + po.getFieldName(), groovyObject);
                } catch (Exception e) {
                    log.error("groovy脚本初始化失败po={}", JSON.toJSONString(po), e);
                }
            }
        }

        //4.赋值
        sourceMap = _sourceMap;
        groovyMap = _groovyMap;

        Map<String, Object> reMap = new HashMap<>();
        reMap.put("_sourceMap", _sourceMap);
        reMap.put("_groovyMap", _groovyMap);
        return reMap;
    }

    // 平铺数据
    @Override
    public List<Map<String, Object>> flatData(Object value) {
        Map<String, Map<String, Object>> tmpList = new HashMap();
        String strValue = value.toString();
        if (StringUtils.isNotBlank(strValue)) {
            Map<String, Object> map = JSON.parseObject(strValue, Map.class);
            String sourceFlag = (String) map.get(ConfConstant.SOURCE_FLAG);
            map.remove(ConfConstant.SOURCE_FLAG);

            if (StringUtils.isEmpty(sourceFlag)) {
                return null;
            }

            List<MappingConf> mappingConfs = sourceMap.get(sourceFlag);
            for (MappingConf conf : mappingConfs) {
                Map<String, Object> oneTableMap = tmpList.get(conf.getTableName());
                if (oneTableMap == null) {
                    oneTableMap = new HashMap<>();
                    tmpList.put(conf.getTableName(), oneTableMap);
                    oneTableMap.put(ConfConstant.TABLE_NAME, conf.getTableName());
                }

                Object reValue = null;
                if ("1".equals(conf.getDealWay())) {
                    reValue = JSONPath.eval(map, conf.getJsonPath());
                } else if ("0".equals(conf.getDealWay())) {
                    GroovyObject groovyObject = groovyMap.get(sourceFlag + "-" + conf.getFieldName());
                    reValue = GroovyUtil.execute(groovyObject, "parse", strValue);
                }
                if (reValue != null && StringUtils.isNotBlank(reValue.toString())) {
                    oneTableMap.put(conf.getFieldName(), reValue);
                }
            }
        }

        List<Map<String, Object>> reList = new LinkedList<>();
        Collection<Map<String, Object>> reMap = tmpList.values();
        for (Map<String, Object> tmpMap : reMap) {
            if (tmpMap.size() >= 2) {
                reList.add(tmpMap);
            }
        }

        return reList.size() > 0 ? reList : null;
    }

    @Override
    public Integer flatDataAndToKafka(Object value) {
        List<Map<String, Object>> list = flatData(value);
        if (CollectionUtils.isEmpty(list)) {
            return 0;
        }
        int re = 0;
        for (Map<String, Object> item : list) {
            if (sendMsg2FlatJsonKafka(item) > 0) {
                re++;
            }
        }
        return re;
    }

    @Override
    public Integer sendMsg2FlatJsonKafka(Object value) {
        int re = 0;
        if (value != null) {
            try {
                RecordMetadata rms = template.send(ConfConstant.TOPIC_FLAT_JSON, JSON.toJSONString(value)).get().getRecordMetadata();
                log.info("kafak消息发送成功,topic={},partion={},offset={}", rms.topic(), rms.partition(), rms.offset());
                re++;
            } catch (Exception e) {
                e.getMessage();
            }
        }
        return re;
    }
}
